/*
 * Copyright (C) 2021 Baidu, Inc. All Rights Reserved.
 */
package com.chukun.rpc.consumer;

import com.chukun.rpc.core.TinyRpcRequest;
import com.chukun.rpc.core.TinyRpcServiceHelper;
import com.chukun.rpc.core.TinyRpcServiceMetaData;
import com.chukun.rpc.protocol.codec.TinyRpcDecoder;
import com.chukun.rpc.protocol.codec.TinyRpcEncoder;
import com.chukun.rpc.protocol.handler.TinyRpcResponseHandler;
import com.chukun.rpc.protocol.protocol.TinyRpcProtocol;
import com.chukun.rpc.registry.RegistryService;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

/**
 * <p>消费者处理类<p/>
 *
 * @author chukun
 * @version 1.0
 * @since 1.0-SNAPSHOT
 */
@Slf4j
public class TinyRpcConsumer {

    private final Bootstrap bootstrap;

    private final EventLoopGroup eventLoopGroup;

    public TinyRpcConsumer() {
        bootstrap = new Bootstrap();
        eventLoopGroup = new NioEventLoopGroup();

        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline()
                                .addLast(new TinyRpcEncoder())
                                .addLast(new TinyRpcDecoder())
                                .addLast(new TinyRpcResponseHandler());
                    }
                });
    }

    /**
     * 发送请求
     * @param protocol
     * @param registryService
     * @throws Exception
     */
    public void sendRequest(TinyRpcProtocol<TinyRpcRequest> protocol, RegistryService registryService)
            throws Exception {
        TinyRpcRequest request = protocol.getBody();
        Object[] params = request.getParams();
        String serviceKey = TinyRpcServiceHelper.buildServiceKey(request.getClassName(),request.getVersion());

        int hash = params.length > 0 ? params[0].hashCode(): serviceKey.hashCode();
        TinyRpcServiceMetaData metaData = registryService.discovery(serviceKey, hash);
        if (metaData !=null) {
            ChannelFuture future = bootstrap.connect(metaData.getServiceAddr(), metaData.getPort()).sync();
            future.addListener((ChannelFutureListener) channelFuture -> {
                if (channelFuture.isSuccess()) {
                    log.info("Connect rpc server {} on port {} success.", metaData.getServiceAddr(), metaData.getPort());
                } else {
                    log.error("Connect rpc server {} on port {} failed.", metaData.getServiceAddr(), metaData.getPort());
                    future.cause().printStackTrace();
                    eventLoopGroup.shutdownGracefully();
                }
            });

            future.channel().writeAndFlush(protocol);
        }
    }

}
